welo git mirrors
Commit 3a36c367fe0aab76fee96d22f45d31d7633a439b
Parents : cf6010d
Author : Mark Qvist <bc7291552be7a58f361522990465165c>
Signature : T66BB85Valid, signed by author
Date : 2026-07-18T02:46:06+02:00
Improved resource handling
Changes
2 files changed, 112 insertions(+), 126 deletions(-)
Diff
diff --git a/RNS/Link.py b/RNS/Link.py
index 7116c7eae..14008b309 100644
--- a/RNS/Link.py
+++ b/RNS/Link.py
@@ -808,7 +808,7 @@ class Link:
request_data = unpacked_request[2]
if path_hash in self.destination.request_handlers:
- request_handler = self.destination.request_handlers[path_hash]
+ request_handler = self.destination.request_handlers[path_hash]
path = request_handler[0]
response_generator = request_handler[1]
allow = request_handler[2]
@@ -1014,36 +1014,39 @@ class Link:
packet.plaintext = self.decrypt(packet.data)
if packet.plaintext != None:
self.__update_phy_stats(packet, query_shared=True)
-
- if RNS.ResourceAdvertisement.is_request(packet):
- RNS.Resource.accept(packet, callback=self.request_resource_concluded)
- elif RNS.ResourceAdvertisement.is_response(packet):
- request_id = RNS.ResourceAdvertisement.read_request_id(packet)
- for pending_request in self.pending_requests:
- if pending_request.request_id == request_id:
- response_resource = RNS.Resource.accept(packet, callback=self.response_resource_concluded, progress_callback=pending_request.response_resource_progress, request_id = request_id)
- if response_resource != None:
- if pending_request.response_size == None:
- pending_request.response_size = RNS.ResourceAdvertisement.read_size(packet)
- if pending_request.response_transfer_size == None:
- pending_request.response_transfer_size = 0
- pending_request.response_transfer_size += RNS.ResourceAdvertisement.read_transfer_size(packet)
- if pending_request.started_at == None:
- pending_request.started_at = time.time()
- pending_request.response_resource_progress(response_resource)
-
- elif self.resource_strategy == Link.ACCEPT_NONE: pass
- elif self.resource_strategy == Link.ACCEPT_APP:
- if self.callbacks.resource != None:
- try:
- resource_advertisement = RNS.ResourceAdvertisement.unpack(packet.plaintext)
- resource_advertisement.link = self
- if self.callbacks.resource(resource_advertisement): RNS.Resource.accept(packet, self.callbacks.resource_concluded)
- else: RNS.Resource.reject(packet)
- except Exception as e:
- RNS.log("Error while executing resource accept callback from "+str(self)+". The contained exception was: "+str(e), RNS.LOG_ERROR)
- elif self.resource_strategy == Link.ACCEPT_ALL:
- RNS.Resource.accept(packet, self.callbacks.resource_concluded)
+ try:
+ if RNS.ResourceAdvertisement.is_request(packet):
+ if self.destination.request_handlers:
+ RNS.Resource.accept(packet, callback=self.request_resource_concluded)
+ elif RNS.ResourceAdvertisement.is_response(packet):
+ request_id = RNS.ResourceAdvertisement.read_request_id(packet)
+ for pending_request in self.pending_requests:
+ if pending_request.request_id == request_id:
+ response_resource = RNS.Resource.accept(packet, callback=self.response_resource_concluded, progress_callback=pending_request.response_resource_progress, request_id = request_id)
+ if response_resource != None:
+ if pending_request.response_size == None:
+ pending_request.response_size = RNS.ResourceAdvertisement.read_size(packet)
+ if pending_request.response_transfer_size == None:
+ pending_request.response_transfer_size = 0
+ pending_request.response_transfer_size += RNS.ResourceAdvertisement.read_transfer_size(packet)
+ if pending_request.started_at == None:
+ pending_request.started_at = time.time()
+ pending_request.response_resource_progress(response_resource)
+ elif self.resource_strategy == Link.ACCEPT_NONE: pass
+ elif self.resource_strategy == Link.ACCEPT_APP:
+ if self.callbacks.resource != None:
+ try:
+ resource_advertisement = RNS.ResourceAdvertisement.unpack(packet.plaintext)
+ resource_advertisement.link = self
+ if self.callbacks.resource(resource_advertisement): RNS.Resource.accept(packet, self.callbacks.resource_concluded)
+ else: RNS.Resource.reject(packet)
+ except Exception as e:
+ RNS.log("Error while executing resource accept callback from "+str(self)+". The contained exception was: "+str(e), RNS.LOG_ERROR)
+ elif self.resource_strategy == Link.ACCEPT_ALL:
+ RNS.Resource.accept(packet, self.callbacks.resource_concluded)
+ except Exception as e:
+ RNS.log(f"Invalid resource advertisement on {self}: {e}", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
+ self.teardown()
elif packet.context == RNS.Packet.RESOURCE_REQ:
plaintext = self.decrypt(packet.data)
diff --git a/RNS/Resource.py b/RNS/Resource.py
index eff056290..696e3f643 100644
--- a/RNS/Resource.py
+++ b/RNS/Resource.py
@@ -239,7 +239,7 @@ class Resource:
return None
except Exception as e:
- RNS.log("Could not decode resource advertisement, dropping resource", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
+ RNS.log(f"Could not decode resource advertisement, dropping resource: {e}", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
return None
# Create a resource for transmission to a remote destination
@@ -314,30 +314,26 @@ class Resource:
self.input_file = data
elif isinstance(data, bytes):
- data_size = len(data)
+ data_size = len(data)
self.total_size = data_size + self.metadata_size
- resource_data = data
+ resource_data = data
self.total_segments = 1
self.segment_index = 1
self.split = False
- elif data == None:
- pass
+ elif data == None: pass
- else:
- raise TypeError("Invalid data instance type passed to resource initialisation")
+ else: raise TypeError("Invalid data instance type passed to resource initialisation")
if resource_data:
if self.has_metadata: data = self.metadata + resource_data
else: data = resource_data
self.status = Resource.NONE
- self.link = link
- if self.link.mtu:
- self.sdu = self.link.mtu - RNS.Reticulum.HEADER_MAXSIZE - RNS.Reticulum.IFAC_MIN_SIZE
- else:
- self.sdu = link.mdu or Resource.SDU
+ self.link = link
+ if self.link.mtu: self.sdu = self.link.mtu - RNS.Reticulum.HEADER_MAXSIZE - RNS.Reticulum.IFAC_MIN_SIZE
+ else: self.sdu = link.mdu or Resource.SDU
self.max_retries = Resource.MAX_RETRIES
self.max_adv_retries = Resource.MAX_ADV_RETRIES
self.retries_left = self.max_retries
@@ -482,12 +478,12 @@ class Resource:
def hashmap_update_packet(self, plaintext):
if not self.status == Resource.FAILED:
- self.last_activity = time.time()
- self.retries_left = self.max_retries
-
- update = umsgpack.unpackb(plaintext[RNS.Identity.HASHLENGTH//8:])
- self.hashmap_update(update[0], update[1])
+ if self.waiting_for_hmu:
+ self.last_activity = time.time()
+ self.retries_left = self.max_retries
+ update = umsgpack.unpackb(plaintext[RNS.Identity.HASHLENGTH//8:])
+ self.hashmap_update(update[0], update[1])
def hashmap_update(self, segment, hashmap):
if not self.status == Resource.FAILED:
@@ -495,12 +491,16 @@ class Resource:
seg_len = ResourceAdvertisement.HASHMAP_MAX_LEN
hashes = len(hashmap)//Resource.MAPHASH_LEN
for i in range(0,hashes):
- if self.hashmap[i+segment*seg_len] == None:
- self.hashmap_height += 1
+ if self.hashmap[i+segment*seg_len] == None: self.hashmap_height += 1
self.hashmap[i+segment*seg_len] = hashmap[i*Resource.MAPHASH_LEN:(i+1)*Resource.MAPHASH_LEN]
- self.waiting_for_hmu = False
- self.request_next()
+ if hashes < 1:
+ RNS.log("Invalid HMU received, cancelling transfer", RNS.LOG_ERROR)
+ self.cancel()
+
+ else:
+ self.waiting_for_hmu = False
+ self.request_next()
def get_map_hash(self, data):
return RNS.Identity.full_hash(data+self.random_hash)[:Resource.MAPHASH_LEN]
@@ -517,6 +517,14 @@ class Resource:
prepare_thread = threading.Thread(target=self.__prepare_next_segment, daemon=True)
prepare_thread.start()
+ def ensure_link(self):
+ if not self.link or self.link.status != RNS.Link.ACTIVE:
+ RNS.log(f"Invalid link state for {self}, aborting transfer", RNS.LOG_VERBOSE)
+ try: self.cancel()
+ except Exception as e: RNS.log(f"Error while cancelling resource on link-state abort: {e}", RNS.LOG_ERROR)
+ return False
+ else: return True
+
def __advertise_job(self):
self.advertisement_packet = RNS.Packet(self.link, ResourceAdvertisement(self).pack(), context=RNS.Packet.RESOURCE_ADV)
while not self.link.ready_for_new_resource():
@@ -524,6 +532,7 @@ class Resource:
sleep(0.25)
try:
+ if not self.ensure_link(): return
self.advertisement_packet.send()
self.last_activity = time.time()
self.started_transferring = self.last_activity
@@ -541,18 +550,13 @@ class Resource:
self.watchdog_job()
def update_eifr(self):
- if self.rtt == None:
- rtt = self.link.rtt
- else:
- rtt = self.rtt
+ if self.rtt == None: rtt = self.link.rtt
+ else: rtt = self.rtt
- if self.req_data_rtt_rate != 0:
- expected_inflight_rate = self.req_data_rtt_rate*8
+ if self.req_data_rtt_rate != 0: expected_inflight_rate = self.req_data_rtt_rate*8
else:
- if self.previous_eifr != None:
- expected_inflight_rate = self.previous_eifr
- else:
- expected_inflight_rate = self.link.establishment_cost*8 / rtt
+ if self.previous_eifr != None: expected_inflight_rate = self.previous_eifr
+ else: expected_inflight_rate = self.link.establishment_cost*8 / rtt
self.eifr = expected_inflight_rate
if self.link: self.link.expected_rate = self.eifr
@@ -566,8 +570,7 @@ class Resource:
this_job_id = self.__watchdog_job_id
while self.status < Resource.ASSEMBLING and this_job_id == self.__watchdog_job_id:
- while self.watchdog_lock:
- sleep(0.025)
+ while self.watchdog_lock: sleep(0.025)
sleep_time = None
if self.status == Resource.ADVERTISED:
@@ -581,6 +584,7 @@ class Resource:
try:
RNS.log("No part requests received, retrying resource advertisement...", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
self.retries_left -= 1
+ if not self.ensure_link(): return
self.advertisement_packet = RNS.Packet(self.link, ResourceAdvertisement(self).pack(), context=RNS.Packet.RESOURCE_ADV)
self.advertisement_packet.send()
self.last_activity = time.time()
@@ -657,13 +661,13 @@ class Resource:
self.last_part_sent = time.time()
sleep_time = 0.001
- elif self.status == Resource.REJECTED:
+ elif self.status >= Resource.ASSEMBLING:
sleep_time = 0.001
if sleep_time == 0:
RNS.log("Warning! Link watchdog sleep time of 0!", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
if sleep_time == None or sleep_time < 0:
- RNS.log("Timing error, cancelling resource transfer.", RNS.LOG_ERROR)
+ RNS.log(f"Timing error ({sleep_time}/{self.status}), cancelling resource transfer.", RNS.LOG_ERROR)
self.cancel()
if sleep_time != None:
@@ -754,6 +758,7 @@ class Resource:
try:
proof = RNS.Identity.full_hash(self.data+self.hash)
proof_data = self.hash+proof
+ if not self.ensure_link(): return
proof_packet = RNS.Packet(self.link, proof_data, packet_type=RNS.Packet.PROOF, context=RNS.Packet.RESOURCE_PRF)
proof_packet.send()
RNS.Transport.cache(proof_packet, force_cache=True)
@@ -776,8 +781,8 @@ class Resource:
advertise = False,
auto_compress = self.auto_compress_option,
sent_metadata_size = self.metadata_size)
- if self.__progress_callback:
- self.next_segment.progress_callback(self.__progress_callback)
+
+ if self.__progress_callback: self.next_segment.progress_callback(self.__progress_callback)
def validate_proof(self, proof_data):
if not self.status == Resource.FAILED:
@@ -966,6 +971,7 @@ class Resource:
request_packet = RNS.Packet(self.link, request_data, context = RNS.Packet.RESOURCE_REQ)
try:
+ if not self.ensure_link(): return
request_packet.send()
self.last_activity = time.time()
self.req_sent = self.last_activity
@@ -1010,11 +1016,11 @@ class Resource:
for part in requested_parts:
try:
+ if not self.ensure_link(): return
if not part.sent:
part.send()
self.sent_parts += 1
- else:
- part.resend()
+ else: part.resend()
self.last_activity = time.time()
self.last_part_sent = self.last_activity
@@ -1032,8 +1038,7 @@ class Resource:
search_end = self.receiver_min_consecutive_height+ResourceAdvertisement.COLLISION_GUARD_SIZE
for part in self.parts[search_start:search_end]:
part_index += 1
- if part.map_hash == last_map_hash:
- break
+ if part.map_hash == last_map_hash: break
self.receiver_min_consecutive_height = max(part_index-1-Resource.WINDOW_MAX, 0)
@@ -1041,10 +1046,8 @@ class Resource:
RNS.log("Resource sequencing error, cancelling transfer!", RNS.LOG_ERROR)
self.cancel()
return
- else:
- segment = part_index // ResourceAdvertisement.HASHMAP_MAX_LEN
+ else: segment = part_index // ResourceAdvertisement.HASHMAP_MAX_LEN
-
hashmap_start = segment*ResourceAdvertisement.HASHMAP_MAX_LEN
hashmap_end = min((segment+1)*ResourceAdvertisement.HASHMAP_MAX_LEN, len(self.parts))
@@ -1052,10 +1055,16 @@ class Resource:
for i in range(hashmap_start,hashmap_end):
hashmap += self.hashmap[i*Resource.MAPHASH_LEN:(i+1)*Resource.MAPHASH_LEN]
+ if not hashmap:
+ RNS.log("Resource HMU error, cancelling transfer!", RNS.LOG_ERROR)
+ self.cancel()
+ return
+
hmu = self.hash+umsgpack.packb([segment, hashmap])
hmu_packet = RNS.Packet(self.link, hmu, context = RNS.Packet.RESOURCE_HMU)
try:
+ if not self.ensure_link(): return
hmu_packet.send()
self.last_activity = time.time()
except Exception as e:
@@ -1245,40 +1254,30 @@ class ResourceAdvertisement:
@staticmethod
def is_request(advertisement_packet):
adv = ResourceAdvertisement.unpack(advertisement_packet.plaintext)
- if adv.q != None and adv.u:
- return True
- else:
- return False
-
+ if adv.q != None and adv.u: return True
+ else: return False
@staticmethod
def is_response(advertisement_packet):
adv = ResourceAdvertisement.unpack(advertisement_packet.plaintext)
-
- if adv.q != None and adv.p:
- return True
- else:
- return False
-
+ if adv.q != None and adv.p: return True
+ else: return False
@staticmethod
def read_request_id(advertisement_packet):
adv = ResourceAdvertisement.unpack(advertisement_packet.plaintext)
return adv.q
-
@staticmethod
def read_transfer_size(advertisement_packet):
adv = ResourceAdvertisement.unpack(advertisement_packet.plaintext)
return adv.t
-
@staticmethod
def read_size(advertisement_packet):
adv = ResourceAdvertisement.unpack(advertisement_packet.plaintext)
return adv.d
-
def __init__(self, resource=None, request_id=None, is_response=False):
self.link = None
if resource != None:
@@ -1310,29 +1309,14 @@ class ResourceAdvertisement:
# Flags
self.f = 0x00 | self.x << 5 | self.p << 4 | self.u << 3 | self.s << 2 | self.c << 1 | self.e
- def get_transfer_size(self):
- return self.t
-
- def get_data_size(self):
- return self.d
-
- def get_parts(self):
- return self.n
-
- def get_segments(self):
- return self.l
-
- def get_hash(self):
- return self.h
-
- def is_compressed(self):
- return self.c
-
- def has_metadata(self):
- return self.x
-
- def get_link(self):
- return self.link
+ def get_transfer_size(self): return self.t
+ def get_data_size(self): return self.d
+ def get_parts(self): return self.n
+ def get_segments(self): return self.l
+ def get_hash(self): return self.h
+ def is_compressed(self): return self.c
+ def has_metadata(self): return self.x
+ def get_link(self): return self.link
def pack(self, segment=0):
hashmap_start = segment*ResourceAdvertisement.HASHMAP_MAX_LEN
@@ -1342,23 +1326,20 @@ class ResourceAdvertisement:
for i in range(hashmap_start,hashmap_end):
hashmap += self.m[i*Resource.MAPHASH_LEN:(i+1)*Resource.MAPHASH_LEN]
- dictionary = {
- "t": self.t, # Transfer size
- "d": self.d, # Data size
- "n": self.n, # Number of parts
- "h": self.h, # Resource hash
- "r": self.r, # Resource random hash
- "o": self.o, # Original hash
- "i": self.i, # Segment index
- "l": self.l, # Total segments
- "q": self.q, # Request ID
- "f": self.f, # Resource flags
- "m": hashmap
- }
+ dictionary = { "t": self.t, # Transfer size
+ "d": self.d, # Data size
+ "n": self.n, # Number of parts
+ "h": self.h, # Resource hash
+ "r": self.r, # Resource random hash
+ "o": self.o, # Original hash
+ "i": self.i, # Segment index
+ "l": self.l, # Total segments
+ "q": self.q, # Request ID
+ "f": self.f, # Resource flags
+ "m": hashmap }
return umsgpack.packb(dictionary)
-
@staticmethod
def unpack(data):
dictionary = umsgpack.unpackb(data)
@@ -1382,4 +1363,6 @@ class ResourceAdvertisement:
adv.p = True if ((adv.f >> 4) & 0x01) == 0x01 else False
adv.x = True if ((adv.f >> 5) & 0x01) == 0x01 else False
+ if adv.t > Resource.MAX_EFFICIENT_SIZE*3: raise ValueError("Invalid transfer size")
+
return adv
\ No newline at end of file
Served by rngit 1.4.2 - Generated in 0.05s